package cn.jly.flink.operators;

import cn.jly.flink.entity.Employee;
import cn.jly.flink.utils.ConstantUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;

import java.math.BigDecimal;

/**
 * max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键， min 和 minBy 同理。
 *
 * @PackageName cn.jly.flink.operators
 * @ClassName KeyByDemo
 * @Description keyBy算子
 * @Author 姬岚洋
 * @Date 2021/1/8 下午 5:04
 */
public class GroupByDemo {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                new Employee(ConstantUtils.ID_GENERATOR.getAndIncrement(), "tom", BigDecimal.valueOf(10000)),
                new Employee(ConstantUtils.ID_GENERATOR.getAndIncrement(), "lily", BigDecimal.valueOf(20000)),
                new Employee(ConstantUtils.ID_GENERATOR.getAndIncrement(), "james", BigDecimal.valueOf(30000))
        )
                .groupBy(new KeySelector<Employee, Long>() {
                    @Override
                    public Long getKey(Employee employee) throws Exception {
                        return employee.getId();
                    }
                })
                .reduce(new ReduceFunction<Employee>() {
                    @Override
                    public Employee reduce(Employee employee, Employee t1) throws Exception {
                        final Employee resEmp = new Employee();
                        resEmp.setSalary(employee.getSalary().add(t1.getSalary()));
                        return resEmp;
                    }
                })
                .print();
    }
}
